package david.java.flink_sql.streamApiIntegration;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static david.java.flink_sql.common.Constant.OUT_PUT_TABLE;

/**
 * @Description: execution behavior  TODO: to be completed !!!
 * <p>
 * The following example shows how updating tables can be converted.
 * Every result row represents an entry in a changelog with a change flag that can be queried by calling row.getKind() on it.
 * In the example, the second score for Alice creates an update before (-U) and update after (+U) change.
 * @link https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/table/data_stream_api/#converting-between-datastream-and-table
 * @Author: ZhaoDawei
 * @Date: Create in 10:19 上午 2022/3/22
 */
public class S1_StreamTableConversion_3 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStreamSource<Row> dataStream = env.fromElements(
                Row.of("Alice", 12),
                Row.of("Bob", 10),
                Row.of("Alice", 100)
        );

        Table inputTable = tableEnv.fromDataStream(dataStream).as("name", "score");

        tableEnv.createTemporaryView("InputTable", inputTable);


        tableEnv.executeSql(OUT_PUT_TABLE);
        System.out.println(OUT_PUT_TABLE);

        tableEnv.from("InputTable").executeInsert("OutputTable");

        tableEnv.executeSql("INSERT INTO OutputTable SELECT * FROM InputTable");

        // tableEnv.createStatementSet()
        //         .addInsert("OutputTable", tableEnv.from("InputTable"))
        //         .addInsert("OutputTable2", tableEnv.from("InputTable"))
        //         .execute();
        //
        // tableEnv.createStatementSet()
        //         .addInsertSql("INSERT INTO OutputTable SELECT * FROM InputTable")
        //         .addInsertSql("INSERT INTO OutputTable2 SELECT * FROM InputTable")
        //         .execute();

// execute with implicit local sink

        tableEnv.from("OutputTable").execute().print();

        // tableEnv.executeSql("SELECT * FROM InputTable").print();
    }
}
